package com.etl.es.charts.base;

import com.eksframework.commons.core.base.EksAsserter;
import com.eksframework.commons.core.base.EksCheck;
import com.eksframework.commons.core.base.EksConverter;
import com.etl.config.ElasticSearchClient;
import com.google.common.base.Objects;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.BucketOrder;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedStringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.NumericMetricsAggregation;
import org.elasticsearch.search.aggregations.metrics.sum.SumAggregationBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Collections;
import java.util.List;

/**
 * 图表查询
 */
@Slf4j
@Component
public class CmsSearchChart implements ISearchChart {

    /**
     * ES客户端
     */
    @Autowired
    ElasticSearchClient elasticSearchClient;

    /**
     * 图表查询接口
     *
     * @param param 图表参数
     * @return ChartsResult 图表结果
     */
    @Override
    public ChartResult search(ChartParam param) {
        this.mergeAggregationCondition(param);
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        QueryBuilder query = queryBuilder(param);
        sourceBuilder.query(query);
        List<AggregationBuilder> aggregations = aggregationBuilder(param);
        EksAsserter.notEmpty(aggregations);
        aggregations.forEach(sourceBuilder::aggregation);
        List<String> indices = param.getSearchIndices();
        EksAsserter.notEmpty(indices);
        SearchRequest searchRequest = new SearchRequest(indices.toArray(new String[0]));
        searchRequest.source(sourceBuilder);
        SearchResponse searchResponse = elasticSearchClient.search(searchRequest);
        return parseSearchResponse(param, searchResponse);
    }

    /**
     * 合并聚合条件
     *
     * @param param 图表参数
     */
    private void mergeAggregationCondition(ChartParam param) {
        AggregationCondition condition = new AggregationCondition();
        if (EksCheck.isEmpty(param.getChartColumns())) {
            param.setChartColumns(condition.getChartColumns());
        }
        if (EksCheck.isBlank(param.getAggregationName())) {
            param.setAggregationName(condition.getAggregationName());
        }
        if (EksCheck.isBlank(param.getGroupName())) {
            param.setGroupName(condition.getGroupName());
        }
        if (EksCheck.isBlank(param.getOrderField())) {
            param.setOrderField(condition.getOrderField());
        }
        if (EksCheck.isNull(param.getOrderAsc())) {
            param.setOrderAsc(condition.getOrderAsc());
        }
        if (EksCheck.isNull(param.getMaxSize())) {
            param.setMaxSize(condition.getMaxSize());
        }
    }

    /**
     * 组装聚合建造器
     *
     * @param param 图表参数
     * @return AggregationBuilder List 聚合建造器
     */
    public List<AggregationBuilder> aggregationBuilder(ChartParam param) {
        SumAggregationBuilder sumAggregationBuilder = AggregationBuilders.sum(
                param.getAggregationName()).field(param.getAggregationField());
        TermsAggregationBuilder termsAggregationBuilder = AggregationBuilders
                .terms(param.getGroupName()).field(param.getGroupField())
                .subAggregation(sumAggregationBuilder)
                .size(param.getMaxSize());
        if (Objects.equal(param.getOrderField(), param.getAggregationName())) {
            termsAggregationBuilder.order(BucketOrder.aggregation(
                    param.getAggregationName(), param.getOrderAsc()));
        } else {
            termsAggregationBuilder.order(BucketOrder.key(param.getOrderAsc()));
        }
        return Collections.singletonList(termsAggregationBuilder);
    }

    /**
     * 解释查询响应
     *
     * @param param          图表参数
     * @param searchResponse 查询响应
     * @return ChartsResult 图表结果
     */
    public ChartResult parseSearchResponse(ChartParam param, SearchResponse searchResponse) {
        ParsedStringTerms stringTerms = searchResponse.getAggregations().get(param.getGroupName());
        List<? extends Terms.Bucket> buckets = stringTerms.getBuckets();
        ChartBuilder chartBuilder = ChartBuilder.of(param.getChartColumns());
        for (Terms.Bucket bucket : buckets) {
            NumericMetricsAggregation.SingleValue agg = bucket.getAggregations()
                    .get(param.getAggregationName());
            chartBuilder.addValues(bucket.getKeyAsString(), EksConverter.of(agg.value()).floatValue(2));
        }
        return chartBuilder.toChartsResult();
    }
}
